package com.bigdata.mapreduce.topn;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;
import java.net.URI;

public class TopN {

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        // 1.从本地加载配置文件得到配置对象
        Configuration conf = new Configuration(true);


        // 在window下并且是local本地模式执行mr任务，需要加一些配置
        // windows异构平台运行
        conf.set("mapreduce.app-submission.cross-platform", "true");
        //决定了集群运行
        conf.set("mapreduce.framework.name", "local");


        // 2.使用配置对象创建Job
        Job job = Job.getInstance(conf);
        // job.setJar("D:\\bigdata_project\\bigdata_2307\\bigdata_hadoop\\target\\bigdata_hadoop-1.0-SNAPSHOT.jar");


        // 3.设置任务的主启动类，job名称
        job.setJarByClass(TopN.class);
        job.setJobName("TopN");

        // 参数个性化
        GenericOptionsParser parser = new GenericOptionsParser(conf, args);
        String[] argsArr = parser.getRemainingArgs();

        Path inputPath = new Path(argsArr[0]);
        Path outputPath = new Path(argsArr[2]);
        TextInputFormat.addInputPath(job, inputPath);

        // 添加缓存路径
        job.addCacheFile(URI.create(argsArr[1]));

        // 判断目录是否存在，如果存在的话，把目录删除
        if (outputPath.getFileSystem(conf).exists(outputPath)) {
            outputPath.getFileSystem(conf).delete(outputPath, true);
        }

        TextOutputFormat.setOutputPath(job, outputPath);

        // 5.设置Mapper类、Mapper的输出key、value类型
        job.setMapperClass(TopnMapper.class);
        // key需要具备比较的功能
        job.setMapOutputKeyClass(Tkey.class);
        job.setMapOutputValueClass(IntWritable.class);

        // 6.设置排序比较器、分区器、分组比较器
        job.setSortComparatorClass(TSortComparator.class);

        job.setPartitionerClass(TPartitioner.class);

        job.setGroupingComparatorClass(TGroupingComparator.class);


        // 7.设置reducer类，reduce输出类型（可选）
        job.setReducerClass(TopNReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 默认reduce task的数量是1个,可以自定义的指定
        // job.setNumReduceTasks(2);

        // 8.提交任务, true 表示监控任务
        job.waitForCompletion(true);
    }
}
